package com.credithc.signature.job.mq;

import com.credithc.eagleeye.plugins.thread.EagleEyeRunnable;
import com.credithc.signature.common.ro.callback.CallbackReq;
import com.credithc.signature.common.utils.MessageFilterUtil;
import com.credithc.signature.job.ro.JobRequest;
import com.credithc.signature.job.service.SignatureService;
import com.credithc.signature.job.utils.ProtobufUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @desc: <br/>
 * @author: Jingbo Hu<br/>
 * @time：2021年06⽉25⽇<br/>
 * @version：v1.0.0<br>
 */
@Slf4j
@Component
public class JobConsumer implements ChannelAwareMessageListener {

    @Autowired
    SignatureService signatureService;


    ThreadPoolExecutor threadPoolExecutor;

    public JobConsumer() {
        this.threadPoolExecutor = new ThreadPoolExecutor(5, 10, 30, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(500), new ThreadPoolExecutor.CallerRunsPolicy());
    }

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        JobRequest jobRequest = null;
        try {
            jobRequest = ProtobufUtil.deserializer(message.getBody(), JobRequest.class);
            this.threadPoolExecutor.execute(new EagleEyeRunnable(new JobTask(jobRequest)));
        } catch (Throwable e) {
            if (jobRequest != null) {
                log.error("－消息－:requestSn={},消息处理失败重新发回MQ", jobRequest.getRequestSn());
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            } else {
                log.error("消息处理失败丢弃消息，原因：消息为空");
            }
        } finally {
            try {
                // 确认消息成功接收
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (Throwable e) {
                // 不做处理
                log.error("");
            }
        }
    }

    /**
     * 多线程任务队列
     */
    class JobTask implements Runnable, Serializable {

        JobRequest jobRequest;

        public JobTask(JobRequest jobRequest) {
            this.jobRequest = jobRequest;
        }

        @Override
        public void run() {
            signatureService.process(jobRequest);
        }
    }

    /**
     * 过滤错误信息中的特殊字符--防止页面查询出错
     *
     * @param callbackReq
     */
    private void errorMessageFilter(CallbackReq callbackReq) {
        //过滤特殊字符
        String errorMsg = callbackReq.getErrorMsg();
        errorMsg = MessageFilterUtil.filter(errorMsg);
        callbackReq.setErrorMsg(errorMsg);
        //限制异常错误信息长度
        errorMsg = MessageFilterUtil.limitLength(errorMsg);
        callbackReq.setErrorMsg(errorMsg);
    }
}

